package com.toutiao.issue

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{HashMap => HMap}
import scala.collection.mutable.{HashSet => HSet}

import java.util.Date

import org.apache.log4j.Logger

import com.toutiao.AppProp

import com.toutiao.score.Scorer
import com.toutiao.score.Measure


class SimTitleModel(
		source: DocTitleStore,
		es: TopicStore,
		scorer: Scorer,
		minV:Double,minS:Double,
		locationDate:String,
		N:Int // 天
    ) {
  
  private
  val _log = Logger.getLogger(this.getClass.getName)
  
  
  private
  val M1 = new HMap[String,Long]
  private
  val M2 = new HMap[Long,String]
  private
  var N1 = 0
  private
  var N2 = 0
  
  
  val Data = source
  
  private
  val Ts = new TimeArray
  
  def maxTime = Ts.maxTime
  def minTime = Ts.minTime
  
  val mData = new SimpleCollector(minV)
  
  _log.info("WAND S"+minV)
  
  val ES = es
  
  def conv(m:DocTitle) = {
    val name = m.name
    val title = m.title
    val t = m.date
    val vs = scorer.score(title)
    new Document(name,vs,title,t)
  }
  
  def processOne(m:Document) = synchronized {
    if (!M1.contains(m.name)){
      N2 += 1
      M1(m.name) = N2
      M2(N2) = m.name
      Ts.insert(m.name, m.date)
      val r = mData.process(M1, M2, m, minS)
      r
    }
    else {
      ""
    }
  }
  
  def process(ms:Array[Document]):Array[String] = {
    val ns = ms.map(e => processOne(e))
    ns
  }
  
  def saveOne(n:String) {
    if (mData.isIn(n)){
      ES.insert(mData(n))
    }
  }
  
  def save(ns:Array[String]) {
    val ms = ns.filter(mData.isIn).map(mData(_))
    ES.insert(ms)
  }
  
  def remove(T:Date) = synchronized {
    val i = Ts.before(T)
    if (i < 1){
      0
    }
    else {
      val ms = Ts.start(i)
      _log.info("remove "+ms.size)
      for (m <- ms if M1.contains(m._1)){
        val name = m._1
        val index = M1(name)
        if (mData.isIn(name)){
          mData.remove(name,index)
        }
        M1 -= name
        M2 -= index
      }
      N1 += ms.size
      Ts.remove(i)
      ms.size
    }
  }
  
  private
  def _init(xs:Array[Topic]) = {
    for (x <- xs if x.ms.size > 0){
      val us = x.ms
      for (u <- us){
        N2 += 1
        M1(u) = N2
        M2(N2) = u
        Ts.insert(u,x.ct)
      }
      mData.insertData(x.name, x)
      mData.insertIndex(M1(x.name), x.title) // 应该合并到mData里去
    }
  }
  
  def init = {
    val current = SimpleDate.loadDate(locationDate)
    val T1 = SimpleDate.increment(current, -N*60)
    val T2 = SimpleDate.increment(current, 60)
    _log.info(SimpleDate.format(T1)+" "+SimpleDate.format(T2))
    
    val xs = ES.retrieve(T1,T2)
    
    _init(xs)
    
    _log.info("MAX_ID "+N2)
    _log.info("MIN_ID "+N1)
  }
  
  // 批处理运行
  def run(m:Int) = synchronized {
    var current = SimpleDate.loadDate(locationDate)
    var eT = Data.getMaxTime("fetch_time")
    
    while (1 == 1) {
      _log.info(SimpleDate.format(eT))
      val x = SimpleDate.increment(current, m)
      if (x.getTime <= eT.getTime) {
        _log.info("Current Date "+SimpleDate.format(current))
        _log.info("End Date "+SimpleDate.format(eT))
        val T1 = current
        val T2 = SimpleDate.increment(T1, m)
        
        val ms = Data.retrieve(T1,T2).map(conv)
        val ns = process(ms)
        
        _log.info("New Event "+ns.size)
        _log.info("Save ... "+ns.size)
        if (ns.size > 0)
          save(ns)
        _log.info("Save Done")
        
        val T = SimpleDate.increment(current, -60*N)
        _log.info(SimpleDate.format(Ts.maxTime))
        _log.info(SimpleDate.format(T))
        _log.info(SimpleDate.format(Ts.minTime))
        val n = remove(T)
        
        _log.info("Remove News "+n)
        _log.info("Current Size "+mData.size)
        _log.info("Current Doc Size "+M1.size)
        _log.info("MAX_ID "+N2)
        _log.info("MIN_ID "+N1)
        val now = SimpleDate.current
      
        current = 
          if (Ts.maxTime.getTime > current.getTime) 
            if (Ts.maxTime.getTime <= now.getTime)
              Ts.maxTime
            else
              now
          else T2
          
        SimpleDate.saveDate(current,locationDate)  
      }
      else {
        val T = SimpleDate.current
        _log.info("Sleep Here "+SimpleDate.format(T))
        Thread.sleep(30*1000)
        eT = Data.getMaxTime("fetch_time")        
      }
      
    }
  }
}

object SimTitleProcess {
  
  def run(prop:AppProp) {
    val source = new DocTitleStore(
    					prop("ES_DOC_NAME"),
    					prop("ES_DOC_URI"),
    					prop("ES_DOC_PORT").toInt,
    					prop("ES_DOC_INDEX"),
    					prop("ES_DOC_TYPE"))
    val es = new TopicStore(
    					prop("ES_TOPIC_NAME"),
    					prop("ES_TOPIC_URI"),
    					prop("ES_TOPIC_PORT").toInt,
    					prop("ES_TOPIC_INDEX"),
    					prop("ES_TOPIC_TYPE"))
    
    val locationDate = prop("LOCATION_DATE")
    val N = prop("TIME_SIZE").toInt
    val minV = prop("MIN_V").toDouble
    val minS = prop("MIN_S").toDouble
 
    val dict = Measure.get(prop("LOCATION_MEASURE"))
    val n = prop("DOC_NUM").toInt
    val scorer = new Scorer(dict,n)
    
    val m = new SimTitleModel(source,es,scorer,minV,minS,locationDate,N)
    m.init
    val K = prop("TIME_STEP").toInt
    m.run(K)
  }
}